/*
 * @Author: gonglf
 * @Date: 2022-06-11 11:26:28
 * @Last Modified by:   gonglf
 * @Last Modified time: 2022-06-11 11:26:28
 */
package brokers

import (
	"log"
	"sync"
	"time"

	"github.com/gomodule/redigo/redis"
)

type redisBroker struct {
	pool            *redis.Pool //redis连接池从外面传进来，由外面启动和关闭，这里只做业务逻辑
	stopChan        chan struct{}
	consumingWG     sync.WaitGroup //一组任务
	processingWG    sync.WaitGroup //正在处理的任务
	concurrency     int            //并发数量
	jobChan         chan []byte    //用来存放等待处理的任务数据，生产者
	workPoolProduce chan struct{}  //用来控制有多少个协程（生产者，消费者，worker）在工作，这个工作可以是：从redis中取出任务
	workPoolConsume chan struct{}  //用来控制有多少个协程（生产者，消费者，worker）在工作，这个工作可以是：worker消费任务
}

func NewRedisBroker(pool *redis.Pool, concurrency int) Broker {
	r := &redisBroker{
		pool:            pool,
		stopChan:        make(chan struct{}),
		concurrency:     concurrency,
		jobChan:         make(chan []byte, concurrency),
		workPoolProduce: make(chan struct{}, concurrency),
		workPoolConsume: make(chan struct{}, concurrency),
	}

	return r
}

//开始消费
/**
 * @description:
 * @param {int} concurrency  并发数量
 * @return {*}
 */
func (r *redisBroker) StartConsuming() (bool, error) {

	for i := 0; i < r.concurrency; i++ { //初始化数量
		r.workPoolProduce <- struct{}{}
	}

	r.consumingWG.Add(1)
	go func() {
		defer func() {
			defer r.consumingWG.Done()
		}()
		log.Println("Waiting for messages. To exit press CTRL+C")
		for {
			select {
			case <-r.stopChan:
				return
			case <-r.workPoolProduce:
				select {
				case <-r.stopChan:
					return
				default:
				}
				go func() {
					//消费任务
					task, _ := r.getTask()
					if len(task) > 0 {
						r.processingWG.Add(1)
						log.Println("++++")
						r.jobChan <- task
					}
					r.workPoolProduce <- struct{}{} //返回一个槽
				}()
			}
		}
	}()

	r.consumingWG.Add(1)
	go r.consume()

	return true, nil
}

//各种close
func (r *redisBroker) StopConsuming() {
	close(r.stopChan)
	r.processingWG.Wait()
	close(r.jobChan) //任务消费完了才能close
	r.consumingWG.Wait()
}

func (r *redisBroker) consume() {
	defer func() {
		r.consumingWG.Done()
	}()
	go func() {
		for i := 0; i < r.concurrency; i++ {
			r.workPoolConsume <- struct{}{}
		}
	}()

	for d := range r.jobChan {
		if r.concurrency > 0 { //如果没有并发数量，默认为无穷大
			<-r.workPoolConsume //同时只能最多执行concurrency个任务
		}

		go func(d []byte) {
			defer r.processingWG.Done()
			log.Println("----")
			r.consumeOne(d)

			if r.concurrency > 0 {
				//返回一个槽
				r.workPoolConsume <- struct{}{}
			}
		}(d)
	}
}

// 处理一条数据，通常比较耗时
func (r *redisBroker) consumeOne(delivery []byte) error {
	defer func() { //防止业务处理发声panic
		if err := recover(); err != nil {
			log.Println("err = ", err)
		}
	}()
	//业务逻辑在这里处理
	log.Println("processing:", string(delivery))
	time.Sleep(time.Second * 3)
	return nil
}

func (r *redisBroker) getTask() (result []byte, err error) {
	time.Sleep(time.Second * 1)
	s := "task"
	return []byte(s), nil
}
